package com.atguigu.champter8;

import com.atguigu.beans.LoginEvent;
import com.atguigu.utils.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

public class Flink06_Project_High_Login {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 9999);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
                .readTextFile("datas/input/LoginLog.csv")
                .map(new MapFunction<String, LoginEvent>() {
                    @Override
                    public LoginEvent map(String value) throws Exception {
                        String[] data = value.split(",");
                        return new LoginEvent(Long.valueOf(data[0]), data[1], data[2], Long.valueOf(data[3]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((ele, ts) -> ele.getEventTime())
                )
                .filter(LoginEvent ->LoginEvent.getEventType().equals("fail"))
                .keyBy(LoginEvent::getUserId)
                .countWindow(2, 1)
                .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>() {
                    @Override
                    public void process(Long userId,
                                        Context context,
                                        Iterable<LoginEvent> elements,
                                        Collector<String> out) throws Exception {
                        List<LoginEvent> list = AtguiguUtil.toList(elements);
                        if (list.size() == 2){
                            Long time1 = list.get(0).getEventTime();
                            Long time2 = list.get(1).getEventTime();

                            if (time1 - time2 < 2){
                                out.collect("用户 : "+ userId + "存在恶意登录风险");
                            }
                        }

                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
